package cn.cihon.ubi.qorosauto.storm;

import java.util.Map;

import cn.cihon.ubi.qorosauto.bo.VehicleStandardData;
import cn.cihon.ubi.qorosauto.dao.file.VehicleStandardDataDao;
import cn.cihon.ubi.qorosauto.dao.hadoop.VehicleStandardDataHdfsDao;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;

public class VehicleStandardDataStorageBolt extends BaseRichBolt{

	/**
	 * 对解析后的明文数据进行存储
	 */
	private static final long serialVersionUID = 1L;
	public OutputCollector collerctor;
	private VehicleStandardDataDao vehicleStandardDataDao = null;
	private VehicleStandardDataHdfsDao vehicleStandardDataHdfsDao = null;
	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		this.collerctor = collector;
		this.vehicleStandardDataDao = new VehicleStandardDataDao();
		this.vehicleStandardDataHdfsDao = new VehicleStandardDataHdfsDao();
	}

	@Override
	public void execute(Tuple input) {
		// TODO Auto-generated method stub
		// 从Spout中获取传来的Tuple
		VehicleStandardData stdData = (VehicleStandardData) input.getValue(1);
		
		// 每次tripData更新时把更新数据写到文件
		 this.vehicleStandardDataDao.storeTofile(stdData);
		 
		 //每次tripData更新时把更新数据写到HDFS
	/*	try 
		{
			this.vehicleStandardDataHdfsDao.writeStdDataToHdfs(stdData);
		} catch (IOException e1) {
			e1.printStackTrace();
		}*/
		this.collerctor.ack(input);
	}
	
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare(new Fields("vehiclestanddata"));
	}

}
